package queue

import (
	"context"
	"encoding/json"
	"fmt"
	"sync"
	"time"

	"github.com/streadway/amqp"
	"go.uber.org/zap"
)

// RabbitMQ RabbitMQ客户端
type RabbitMQ struct {
	conn         *amqp.Connection
	channel      *amqp.Channel
	exchangeName string
	queueName    string
	routingKey   string
	exchangeType string
	url          string
	mutex        sync.RWMutex
	consumers    map[string]ConsumerFunc
	reconnecting bool
	ctx          context.Context
	cancel       context.CancelFunc
	wg           sync.WaitGroup
}

// Config RabbitMQ配置
type Config struct {
	Host         string
	Port         int
	Username     string
	Password     string
	VHost        string
	ExchangeName string
	QueueName    string
	RoutingKey   string
	ExchangeType string
	Durable      bool
	AutoDelete   bool
	Exclusive    bool
	NoWait       bool
}

// Message 消息结构
type Message struct {
	ID        string      `json:"id"`
	Type      string      `json:"type"`
	Content   interface{} `json:"content"`
	Timestamp int64       `json:"timestamp"`
	Source    string      `json:"source"`
}

// ConsumerFunc 消费者函数类型
type ConsumerFunc func(msg amqp.Delivery) error

// NewRabbitMQ 创建RabbitMQ客户端
func NewRabbitMQ(config *Config) (*RabbitMQ, error) {
	// 构建连接URL
	url := fmt.Sprintf("amqp://%s:%s@%s:%d/%s",
		config.Username,
		config.Password,
		config.Host,
		config.Port,
		config.VHost,
	)

	// 创建上下文
	ctx, cancel := context.WithCancel(context.Background())

	// 创建RabbitMQ客户端
	mq := &RabbitMQ{
		url:          url,
		exchangeName: config.ExchangeName,
		queueName:    config.QueueName,
		routingKey:   config.RoutingKey,
		exchangeType: config.ExchangeType,
		consumers:    make(map[string]ConsumerFunc),
		ctx:          ctx,
		cancel:       cancel,
	}

	// 连接RabbitMQ服务器
	if err := mq.connect(); err != nil {
		cancel()
		return nil, err
	}

	// 启动重连监控
	go mq.reconnectMonitor()

	return mq, nil
}

// connect 连接RabbitMQ服务器
func (mq *RabbitMQ) connect() error {
	mq.mutex.Lock()
	defer mq.mutex.Unlock()

	// 创建连接
	conn, err := amqp.Dial(mq.url)
	if err != nil {
		return fmt.Errorf("连接RabbitMQ服务器失败: %v", err)
	}

	// 创建通道
	channel, err := conn.Channel()
	if err != nil {
		conn.Close()
		return fmt.Errorf("创建通道失败: %v", err)
	}

	// 声明交换机
	if mq.exchangeName != "" {
		err = channel.ExchangeDeclare(
			mq.exchangeName, // 交换机名称
			mq.exchangeType, // 交换机类型
			true,            // 持久化
			false,           // 自动删除
			false,           // 内部交换机
			false,           // 非等待
			nil,             // 参数
		)
		if err != nil {
			channel.Close()
			conn.Close()
			return fmt.Errorf("声明交换机失败: %v", err)
		}
	}

	// 声明队列
	if mq.queueName != "" {
		_, err = channel.QueueDeclare(
			mq.queueName, // 队列名称
			true,         // 持久化
			false,        // 自动删除
			false,        // 排他性
			false,        // 非等待
			nil,          // 参数
		)
		if err != nil {
			channel.Close()
			conn.Close()
			return fmt.Errorf("声明队列失败: %v", err)
		}

		// 绑定队列到交换机
		if mq.exchangeName != "" {
			err = channel.QueueBind(
				mq.queueName,    // 队列名称
				mq.routingKey,   // 路由键
				mq.exchangeName, // 交换机名称
				false,           // 非等待
				nil,             // 参数
			)
			if err != nil {
				channel.Close()
				conn.Close()
				return fmt.Errorf("绑定队列到交换机失败: %v", err)
			}
		}
	}

	// 设置QoS
	err = channel.Qos(
		1,     // 预取计数
		0,     // 预取大小
		false, // 全局
	)
	if err != nil {
		channel.Close()
		conn.Close()
		return fmt.Errorf("设置QoS失败: %v", err)
	}

	// 保存连接和通道
	mq.conn = conn
	mq.channel = channel

	// 监听连接关闭事件
	go func() {
		<-conn.NotifyClose(make(chan *amqp.Error))
		zap.L().Warn("RabbitMQ连接已关闭，准备重连")
		mq.triggerReconnect()
	}()

	zap.L().Info("RabbitMQ连接成功",
		zap.String("exchange", mq.exchangeName),
		zap.String("queue", mq.queueName),
		zap.String("routing_key", mq.routingKey),
	)

	return nil
}

// reconnectMonitor 重连监控
func (mq *RabbitMQ) reconnectMonitor() {
	for {
		select {
		case <-mq.ctx.Done():
			return
		default:
			if mq.reconnecting {
				mq.mutex.RLock()
				reconnecting := mq.reconnecting
				mq.mutex.RUnlock()

				if reconnecting {
					// 尝试重新连接
					err := mq.connect()
					if err != nil {
						zap.L().Error("RabbitMQ重连失败", zap.Error(err))
						time.Sleep(5 * time.Second) // 等待5秒后重试
						continue
					}

					// 重连成功，恢复所有消费者
					mq.mutex.Lock()
					mq.reconnecting = false
					consumers := make(map[string]ConsumerFunc)
					for tag, consumer := range mq.consumers {
						consumers[tag] = consumer
					}
					mq.mutex.Unlock()

					// 重新启动所有消费者
					for tag, consumer := range consumers {
						err := mq.registerConsumer(tag, consumer)
						if err != nil {
							zap.L().Error("恢复消费者失败", zap.String("tag", tag), zap.Error(err))
						}
					}
				}
			}
			time.Sleep(1 * time.Second)
		}
	}
}

// triggerReconnect 触发重连
func (mq *RabbitMQ) triggerReconnect() {
	mq.mutex.Lock()
	defer mq.mutex.Unlock()

	// 标记为重连中
	mq.reconnecting = true

	// 关闭旧的连接和通道
	if mq.channel != nil {
		mq.channel.Close()
		mq.channel = nil
	}
	if mq.conn != nil {
		mq.conn.Close()
		mq.conn = nil
	}
}

// Publish 发布消息
func (mq *RabbitMQ) Publish(msg interface{}) error {
	mq.mutex.RLock()
	defer mq.mutex.RUnlock()

	if mq.channel == nil {
		return fmt.Errorf("RabbitMQ通道未初始化")
	}

	// 序列化消息
	data, err := json.Marshal(msg)
	if err != nil {
		return fmt.Errorf("序列化消息失败: %v", err)
	}

	// 发布消息
	err = mq.channel.Publish(
		mq.exchangeName, // 交换机名称
		mq.routingKey,   // 路由键
		false,           // 强制
		false,           // 立即
		amqp.Publishing{
			ContentType:  "application/json",
			Body:         data,
			DeliveryMode: amqp.Persistent, // 持久化消息
			Timestamp:    time.Now(),
		},
	)
	if err != nil {
		return fmt.Errorf("发布消息失败: %v", err)
	}

	return nil
}

// PublishWithRoutingKey 使用指定的路由键发布消息
func (mq *RabbitMQ) PublishWithRoutingKey(routingKey string, msg interface{}) error {
	mq.mutex.RLock()
	defer mq.mutex.RUnlock()

	if mq.channel == nil {
		return fmt.Errorf("RabbitMQ通道未初始化")
	}

	// 序列化消息
	data, err := json.Marshal(msg)
	if err != nil {
		return fmt.Errorf("序列化消息失败: %v", err)
	}

	// 发布消息
	err = mq.channel.Publish(
		mq.exchangeName, // 交换机名称
		routingKey,      // 路由键
		false,           // 强制
		false,           // 立即
		amqp.Publishing{
			ContentType:  "application/json",
			Body:         data,
			DeliveryMode: amqp.Persistent, // 持久化消息
			Timestamp:    time.Now(),
		},
	)
	if err != nil {
		return fmt.Errorf("发布消息失败: %v", err)
	}

	return nil
}

// Consume 消费消息
func (mq *RabbitMQ) Consume(consumerTag string, autoAck bool, handler ConsumerFunc) error {
	mq.mutex.Lock()
	defer mq.mutex.Unlock()

	if mq.channel == nil {
		return fmt.Errorf("RabbitMQ通道未初始化")
	}

	// 保存消费者函数
	mq.consumers[consumerTag] = handler

	return mq.registerConsumer(consumerTag, handler)
}

// registerConsumer 注册消费者
func (mq *RabbitMQ) registerConsumer(consumerTag string, handler ConsumerFunc) error {
	if mq.channel == nil {
		return fmt.Errorf("RabbitMQ通道未初始化")
	}

	// 消费消息
	deliveries, err := mq.channel.Consume(
		mq.queueName, // 队列名称
		consumerTag,  // 消费者标签
		false,        // 自动确认
		false,        // 排他性
		false,        // 非本地
		false,        // 非等待
		nil,          // 参数
	)
	if err != nil {
		return fmt.Errorf("注册消费者失败: %v", err)
	}

	// 启动消费者协程
	mq.wg.Add(1)
	go func() {
		defer mq.wg.Done()
		for {
			select {
			case <-mq.ctx.Done():
				return
			case delivery, ok := <-deliveries:
				if !ok {
					zap.L().Warn("消费者通道已关闭", zap.String("consumer", consumerTag))
					return
				}

				// 处理消息
				err := handler(delivery)
				if err != nil {
					zap.L().Error("处理消息失败",
						zap.String("consumer", consumerTag),
						zap.Error(err),
						zap.ByteString("body", delivery.Body),
					)
					// 拒绝消息并重新入队
					delivery.Reject(true)
				} else {
					// 确认消息
					delivery.Ack(false)
				}
			}
		}
	}()

	zap.L().Info("消费者已注册", zap.String("consumer", consumerTag), zap.String("queue", mq.queueName))
	return nil
}

// GetChannel 获取RabbitMQ通道
func (mq *RabbitMQ) GetChannel() (*amqp.Channel, error) {
	mq.mutex.RLock()
	defer mq.mutex.RUnlock()

	if mq.channel == nil {
		return nil, fmt.Errorf("RabbitMQ通道未初始化")
	}

	return mq.channel, nil
}

// Close 关闭RabbitMQ连接
func (mq *RabbitMQ) Close() error {
	mq.mutex.Lock()
	defer mq.mutex.Unlock()

	// 取消上下文
	mq.cancel()

	// 等待所有消费者退出
	mq.wg.Wait()

	// 关闭通道和连接
	if mq.channel != nil {
		mq.channel.Close()
		mq.channel = nil
	}
	if mq.conn != nil {
		mq.conn.Close()
		mq.conn = nil
	}

	zap.L().Info("RabbitMQ连接已关闭")
	return nil
}

// IsConnected 检查是否已连接
func (mq *RabbitMQ) IsConnected() bool {
	mq.mutex.RLock()
	defer mq.mutex.RUnlock()
	return mq.conn != nil && mq.channel != nil && !mq.reconnecting
}

// CreateMessage 创建消息
func CreateMessage(msgType string, content interface{}, source string) *Message {
	return &Message{
		ID:        fmt.Sprintf("%d", time.Now().UnixNano()),
		Type:      msgType,
		Content:   content,
		Timestamp: time.Now().Unix(),
		Source:    source,
	}
}
